package jnpf.message.service.impl;

import cn.hutool.core.util.BooleanUtil;
import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.dynamic.datasource.DynamicRoutingDataSource;
import com.baomidou.dynamic.datasource.annotation.DSTransactional;
import com.baomidou.dynamic.datasource.ds.GroupDataSource;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jnpf.base.Pagination;
import jnpf.base.UserInfo;
import jnpf.base.entity.MessageTemplateEntity;
import jnpf.config.ConfigValueUtil;
import jnpf.util.data.DataSourceContextHolder;
import jnpf.database.util.ConnUtil;
import jnpf.database.util.DataSourceUtil;
import jnpf.message.entity.MessageEntity;
import jnpf.message.entity.MessageReceiveEntity;
import jnpf.message.mapper.MessageMapper;
import jnpf.message.service.MessageService;
import jnpf.message.service.MessagereceiveService;
import jnpf.model.OnlineUserModel;
import jnpf.model.OnlineUserProvider;
import jnpf.util.*;
import jnpf.util.context.SpringContext;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * 消息实例
 *
 * @author JNPF开发平台组
 * @version V3.1.0
 * @copyright 引迈信息技术有限公司
 * @date 2019年9月27日 上午9:18
 */
@Service
@Slf4j
public class MessageServiceImpl extends ServiceImpl<MessageMapper, MessageEntity> implements MessageService {

    @Autowired
    private UserProvider userProvider;
    @Autowired
    private ConfigValueUtil configValueUtil;
    @Autowired
    private DataSourceUtil dataSourceUtils;
    @Autowired
    private MessagereceiveService messagereceiveService;

    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private DataSource dataSource;

    @Override
    public List<MessageEntity> getNoticeList(Pagination pagination) {
        // 定义变量判断是否需要使用修改时间倒序
        boolean flag = false;
        QueryWrapper<MessageEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageEntity::getType, 1);
        //关键词（消息标题）
        if (!StringUtil.isEmpty(pagination.getKeyword())) {
            flag = true;
            queryWrapper.lambda().like(MessageEntity::getTitle, pagination.getKeyword());
        }
        //默认排序
        queryWrapper.lambda().orderByAsc(MessageEntity::getEnabledMark).orderByDesc(MessageEntity::getCreatorTime);
        if (flag) {
            queryWrapper.lambda().orderByDesc(MessageEntity::getLastModifyTime);
        }
        queryWrapper.lambda().select(MessageEntity::getId, MessageEntity::getCreatorUser, MessageEntity::getEnabledMark, MessageEntity::getLastModifyTime, MessageEntity::getTitle, MessageEntity::getCreatorTime, MessageEntity::getLastModifyUserId);
        Page<MessageEntity> page = new Page<>(pagination.getCurrentPage(), pagination.getPageSize());
        IPage<MessageEntity> iPage = this.page(page, queryWrapper);
        return pagination.setData(iPage.getRecords(), page.getTotal());
    }

    @Override
    public List<MessageEntity> getNoticeList() {
        QueryWrapper<MessageEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageEntity::getType, 1).eq(MessageEntity::getEnabledMark, 1);
        queryWrapper.lambda().orderByAsc(MessageEntity::getSortCode);
        return this.baseMapper.selectList(queryWrapper);
    }

    @Override
    public List<MessageEntity> getDashboardNoticeList() {
        List<MessageReceiveEntity> list = new ArrayList<>(16);
        QueryWrapper<MessageEntity> queryWrapper1 = new QueryWrapper<>();
        if (!userProvider.get().getIsAdministrator()) {
            // 判断哪些消息是自己接收的
            QueryWrapper<MessageReceiveEntity> queryWrapper = new QueryWrapper<>();
            queryWrapper.lambda().eq(MessageReceiveEntity::getUserId, userProvider.get().getUserId());
            list = messagereceiveService.list(queryWrapper);
            List<String> collect = new ArrayList<>(16);
            if (list.size() > 0) {
                collect = list.stream().map(MessageReceiveEntity::getMessageId).collect(Collectors.toList());
            } else {
                collect.add("");
            }
            queryWrapper1.lambda().in(MessageEntity::getId, collect);
        }
        // 得到公告内容
        queryWrapper1.lambda().eq(MessageEntity::getType, 1).eq(MessageEntity::getEnabledMark, 1);
        queryWrapper1.lambda().orderByAsc(MessageEntity::getSortCode).orderByDesc(MessageEntity::getCreatorTime);
        Page<MessageEntity> page = new Page<>(1, 20);
        IPage<MessageEntity> iPage = this.page(page, queryWrapper1);
        return iPage.getRecords();
    }

    @Override
    public List<MessageEntity> getMessageList(Pagination pagination, String type) {
        return getMessageList(pagination, type, null);
    }

    @Override
    public List<MessageEntity> getMessageList(Pagination pagination, String type, String user) {
        String userId = StringUtil.isEmpty(user) ? userProvider.get().getUserId() : user;
        Map<String, Object> map = new HashMap<>(16);
        map.put("userId", userId);
        //关键词（消息标题）
        String keyword = pagination.getKeyword();
        if (!StringUtil.isEmpty(keyword)) {
            map.put("keyword", "%" + keyword + "%");
        }
        //消息类别
        if (!StringUtil.isEmpty(type)) {
            map.put("type", Integer.valueOf(type));
        }
        List<MessageEntity> lists = this.baseMapper.getMessageList(map);
        return pagination.setData(PageUtil.getListPage((int) pagination.getCurrentPage(), (int) pagination.getPageSize(), lists), lists.size());
    }

    @Override
    public List<MessageEntity> getMessageList(Pagination pagination) {
        return this.getMessageList(pagination, null);
    }

    @Override
    public MessageEntity getinfo(String id) {
        QueryWrapper<MessageEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageEntity::getId, id);
        return this.getOne(queryWrapper);
    }

    @Override
    public MessageEntity getInfoDefault(int type) {
        QueryWrapper<MessageEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageEntity::getType, type).eq(MessageEntity::getEnabledMark, 1);
        if (type == 1) {
            queryWrapper.lambda().orderByDesc(MessageEntity::getCreatorTime);
        } else {
            queryWrapper.lambda().orderByDesc(MessageEntity::getLastModifyTime);
        }
        // 只查询id
        queryWrapper.lambda().select(MessageEntity::getId, MessageEntity::getTitle, MessageEntity::getCreatorTime);
        List<MessageEntity> list = this.page(new Page<>(1, 1, false), queryWrapper).getRecords();
        MessageEntity entity = new MessageEntity();
        if (list.size() > 0) {
            entity = list.get(0);
        }
        return entity;
    }

    @Override
    @DSTransactional
    public void delete(MessageEntity entity) {
        this.removeById(entity.getId());
        QueryWrapper<MessageReceiveEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageReceiveEntity::getMessageId, entity.getId());
        messagereceiveService.remove(queryWrapper);
    }

    @Override
    public void create(MessageEntity entity) {
        entity.setId(RandomUtil.uuId());
        entity.setBodyText(entity.getBodyText());
        entity.setType(1);
        entity.setEnabledMark(0);
        entity.setCreatorUser(userProvider.get().getUserId());
        this.save(entity);
    }

    @Override
    public boolean update(String id, MessageEntity entity) {
        entity.setId(id);
        entity.setBodyText(entity.getBodyText());
        entity.setCreatorUser(userProvider.get().getUserId());
        return this.updateById(entity);
    }

    @Override
    public MessageReceiveEntity messageRead(String messageId) {
        String userId = userProvider.get().getUserId();
        QueryWrapper<MessageReceiveEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageReceiveEntity::getUserId, userId).eq(MessageReceiveEntity::getMessageId, messageId);
        MessageReceiveEntity entity = messagereceiveService.getOne(queryWrapper);
        if (entity != null) {
            entity.setIsRead(1);
            entity.setReadCount(entity.getReadCount() == null ? 1 : entity.getReadCount() + 1);
            entity.setReadTime(new Date());
            messagereceiveService.updateById(entity);
        }
        return entity;
    }

    @Override
    @DSTransactional
    public void messageRead() {
        String userId = userProvider.get().getUserId();
        QueryWrapper<MessageReceiveEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageReceiveEntity::getUserId, userId).eq(MessageReceiveEntity::getIsRead, 0);
        List<MessageReceiveEntity> entitys = messagereceiveService.list(queryWrapper);
        if (entitys.size() > 0) {
            for (MessageReceiveEntity entity : entitys) {
                entity.setIsRead(1);
                entity.setReadCount(entity.getReadCount() == null ? 1 : entity.getReadCount() + 1);
                entity.setReadTime(new Date());
                messagereceiveService.updateById(entity);
            }
        }
    }

    @Override
    @DSTransactional
    public void deleteRecord(List<String> messageIds) {
        QueryWrapper<MessageReceiveEntity> queryWrapper = new QueryWrapper<>();
        // 通过id删除无需判断接收人
        // .eq(MessageReceiveEntity::getUserId, userId)
        queryWrapper.lambda().in(MessageReceiveEntity::getMessageId, messageIds);
        messagereceiveService.remove(queryWrapper);
        // 删除消息表
        QueryWrapper<MessageEntity> queryWrapper1 = new QueryWrapper<>();
        queryWrapper1.lambda().in(MessageEntity::getId, messageIds);
        this.remove(queryWrapper1);
    }

    @Override
    public int getUnreadCount(String userId) {
        QueryWrapper<MessageReceiveEntity> queryWrapper = new QueryWrapper<>();
        queryWrapper.lambda().eq(MessageReceiveEntity::getUserId, userId).eq(MessageReceiveEntity::getIsRead, 0);
        return (int) messagereceiveService.count(queryWrapper);
    }

    @Override
    public int getUnreadNoticeCount(String userId) {
        int result = this.baseMapper.getUnreadNoticeCount(userId);
        return result;
    }

    @Override
    public int getUnreadMessageCount(String userId) {
        int result = this.baseMapper.getUnreadMessageCount(userId);
        return result;
    }

    @Override
    @DSTransactional
    public void sentNotice(List<String> toUserIds, MessageEntity entity) {
        UserInfo userInfo = userProvider.get();
        entity.setEnabledMark(1);
        entity.setLastModifyTime(new Date());
        entity.setLastModifyUserId(userInfo.getUserId());
        this.updateById(entity);
        // 存到redis中的key对象
        List<String> idList = new ArrayList<>();
        // 存到redis，生成Redis的key
        Callable executeInsert = () -> {
            try {
                executeInsert(toUserIds, idList);
            } catch (Exception e) {
                entity.setEnabledMark(0);
                entity.setLastModifyTime(new Date());
                entity.setLastModifyUserId(userInfo.getUserId());
                this.updateById(entity);
                return false;
            }
            return true;
        };
        ThreadPoolTaskExecutor threadPoolExecutor = SpringContext.getBean(ThreadPoolTaskExecutor.class);
        Future<Boolean> submit = threadPoolExecutor.submit(executeInsert);
        try {
            if (submit.get()) {
                // 执行发送公告操作
                Runnable runnable = () -> executeBatch(idList, entity, userInfo);
                threadPoolExecutor.submit(runnable);
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 数据存到redis中
     *
     * @param toUserIds 接受者id
     */
    private void executeInsert(List<String> toUserIds, List<String> idList) throws Exception {
        List<String> key = new ArrayList<>();
        try {
            int frequency = 10000;
            int count = toUserIds.size() / frequency + 1;
            for (int i = 0; i < count; i++) {
                // 生成redis的key
                String cacheKey = RandomUtil.uuId() + toUserIds.get(i);
                // 存到redis
                int endSize = Math.min(((i + 1) * frequency), toUserIds.size());
                redisUtil.insert(cacheKey, toUserIds.subList(i * frequency, endSize));
                key.add(cacheKey);
            }
        } catch (Exception e) {
            key.clear();
            throw new Exception();
        }
        idList.addAll(key);
    }

    /**
     * 执行发送操作
     *
     * @param idList   存到redis中的key
     * @param entity
     * @param userInfo
     */
    private void executeBatch(List<String> idList, MessageEntity entity, UserInfo userInfo) {
        List<String> toUserIdList = null;
        String cacheKey = null;
        try {
            String dbName = configValueUtil.isMultiTenancy() ?
                    userProvider.get().getTenantDbConnectionString() : dataSourceUtils.getDbName();
            @Cleanup Connection conn = null;
            if (StringUtil.isNotEmpty(userInfo.getTenantId()) && "true".equals(userInfo.getTenantDbConnectionString())) {
                DynamicRoutingDataSource dynamicRoutingDataSource = (DynamicRoutingDataSource) dataSource;
                if (dynamicRoutingDataSource.getGroupDataSources().containsKey(DataSourceContextHolder.getDatasourceId() + "-master")) {
                    GroupDataSource groupDataSource = dynamicRoutingDataSource.getGroupDataSources().get(DataSourceContextHolder.getDatasourceId() + "-master");
                    if (groupDataSource.getDataSourceMap().size() > 0) {
                        String key = "";
                        for (String string : groupDataSource.getDataSourceMap().keySet()) {
                            key = string;
                            break;
                        }
                        DruidDataSource dataSource = (DruidDataSource) groupDataSource.getDataSourceMap().get(key);
                        conn = ConnUtil.getConn(dataSource.getUsername(), dataSource.getPassword(), dataSource.getUrl());
                    }
                }
            } else {
                conn = ConnUtil.getConn(dataSourceUtils, dbName);
            }
            if (configValueUtil.isMultiTenancy() && "PostgreSQL".equals(dataSourceUtils.getDbType())) {
                if (!"true".equals(userInfo.getTenantDbConnectionString())) {
                    dbName = dbName + ".";
                }
            } else {
                dbName = "";
            }
            String sql = "INSERT INTO " + dbName + "base_messagereceive(F_ID,F_MESSAGEID,F_USERID,F_ISREAD,F_BODYTEXT)  VALUES (?,?,?,?,?)";
            for (int i = 0; i < idList.size(); i++) {
                @Cleanup PreparedStatement pstm = null;

                pstm = conn.prepareStatement(sql);
                conn.setAutoCommit(false);
                cacheKey = idList.get(i);
                List<List<String>> list = (List) redisUtil.get(cacheKey, 0, -1);
                toUserIdList = list.get(0);
                for (String toUserId : toUserIdList) {
                    pstm.setString(1, RandomUtil.uuId());
                    pstm.setString(2, entity.getId());
                    pstm.setString(3, toUserId);
                    pstm.setInt(4, 0);
                    pstm.setString(5, entity.getBodyText());
                    pstm.addBatch();
                }
                long time = System.currentTimeMillis();
                log.info("开始提交事务，时间：" + time);
                pstm.executeBatch();
                conn.commit();
                long endTime = System.currentTimeMillis();
                log.info("事务提交完成：" + endTime + "耗时：" + (endTime - time));
                pstm.close();
                conn.close();
                //消息推送 - PC端
                for (int j = 0; j < toUserIdList.size(); j++) {
                    for (OnlineUserModel item : OnlineUserProvider.getOnlineUserList()) {
                        if (toUserIdList.get(j).equals(item.getUserId()) && userInfo.getTenantId().equals(item.getTenantId())) {
                            JSONObject map = new JSONObject();
                            map.put("method", "messagePush");
                            map.put("unreadNoticeCount", 1);
                            map.put("messageType", 1);
                            map.put("userId", userInfo.getTenantId());
                            map.put("toUserId", toUserIdList);
                            map.put("title", entity.getTitle());
                            map.put("messageDefaultTime", entity.getLastModifyTime() != null ? entity.getLastModifyTime().getTime() : null);
                            OnlineUserProvider.sendMessage(item, map);
                        }
                    }
                }
                redisUtil.remove(cacheKey);
            }
        } catch (Exception e) {
            log.error(e.getMessage());
            log.error("发送消息发生错误，Key值：" + cacheKey + "，发送人：" + userInfo.getUserId());
        }

    }


    @Override
    public void sentMessage(List<String> toUserIds, String title) {
        this.sentMessage(toUserIds, title, null);
    }

    @Override
    @DSTransactional
    public void sentMessage(List<String> toUserIds, String title, String bodyText) {
        UserInfo userInfo = userProvider.get();
        MessageEntity entity = new MessageEntity();
        entity.setTitle(title);
        entity.setBodyText(bodyText);
        entity.setId(RandomUtil.uuId());
        entity.setType(2);
        entity.setCreatorUser(userInfo.getUserId());
        entity.setCreatorTime(new Date());
        entity.setLastModifyTime(entity.getCreatorTime());
        entity.setLastModifyUserId(entity.getCreatorUser());
        List<MessageReceiveEntity> receiveEntityList = new ArrayList<>();
        for (String item : toUserIds) {
            MessageReceiveEntity messageReceiveEntity = new MessageReceiveEntity();
            messageReceiveEntity.setId(RandomUtil.uuId());
            messageReceiveEntity.setMessageId(entity.getId());
            messageReceiveEntity.setUserId(item);
            messageReceiveEntity.setIsRead(0);
            messageReceiveEntity.setBodyText(bodyText);
            receiveEntityList.add(messageReceiveEntity);
        }
        this.save(entity);
        for (MessageReceiveEntity messageReceiveEntity : receiveEntityList) {
            messagereceiveService.save(messageReceiveEntity);
        }
        //消息推送 - PC端
        for (int i = 0; i < toUserIds.size(); i++) {
            for (OnlineUserModel item : OnlineUserProvider.getOnlineUserList()) {
                if (toUserIds.get(i).equals(item.getUserId()) && userInfo.getTenantId().equals(item.getTenantId())) {
                    JSONObject map = new JSONObject();
                    map.put("method", "messagePush");
                    map.put("unreadNoticeCount", 1);
                    map.put("messageType", 2);
                    map.put("userId", userInfo.getTenantId());
                    map.put("toUserId", toUserIds);
                    map.put("title", entity.getTitle());
                    map.put("messageDefaultTime", entity.getLastModifyTime() != null ? entity.getLastModifyTime().getTime() : null);
                    OnlineUserProvider.sendMessage(item, map);
                }
            }
        }
    }

    @Override
    @DSTransactional
    public void sentMessage(List<String> toUserIds, String title, String bodyText, Map<String, String> contentMsg, UserInfo userInfo) {
        MessageEntity entity = new MessageEntity();
        entity.setTitle(title);
        entity.setBodyText(bodyText);
        entity.setId(RandomUtil.uuId());
        entity.setType(2);
        entity.setEnabledMark(1);
        entity.setCreatorUser(userInfo.getUserId());
        entity.setCreatorTime(new Date());
        entity.setLastModifyTime(entity.getCreatorTime());
        entity.setLastModifyUserId(entity.getCreatorUser());
        List<MessageReceiveEntity> receiveEntityList = new ArrayList<>();
        for (String item : toUserIds) {
            MessageReceiveEntity messageReceiveEntity = new MessageReceiveEntity();
            messageReceiveEntity.setId(RandomUtil.uuId());
            messageReceiveEntity.setMessageId(entity.getId());
            messageReceiveEntity.setUserId(item);
            messageReceiveEntity.setIsRead(0);
            String msg = contentMsg.get(item) != null ? contentMsg.get(item) : "{}";
            messageReceiveEntity.setBodyText(msg);
            receiveEntityList.add(messageReceiveEntity);
        }
        this.save(entity);
        for (MessageReceiveEntity messageReceiveEntity : receiveEntityList) {
            messagereceiveService.save(messageReceiveEntity);
        }
        //消息推送 - PC端
        for (int i = 0; i < toUserIds.size(); i++) {
            for (OnlineUserModel item : OnlineUserProvider.getOnlineUserList()) {
                if (toUserIds.get(i).equals(item.getUserId()) && userInfo.getTenantId().equals(item.getTenantId())) {
                    JSONObject map = new JSONObject();
                    map.put("method", "messagePush");
                    map.put("unreadNoticeCount", 1);
                    map.put("messageType", 2);
                    map.put("userId", userInfo.getTenantId());
                    map.put("toUserId", toUserIds);
                    map.put("title", entity.getTitle());
                    map.put("messageDefaultTime", entity.getLastModifyTime() != null ? entity.getLastModifyTime().getTime() : null);
                    OnlineUserProvider.sendMessage(item, map);
                }
            }
        }
    }

    @Override
    @DSTransactional
    public void sentMessage(List<String> toUserIds, String title, String bodyText, UserInfo userInfo) {
        MessageEntity entity = new MessageEntity();
        entity.setTitle(title);
        entity.setBodyText(bodyText);
        entity.setId(RandomUtil.uuId());
        entity.setType(2);
        entity.setCreatorUser(userInfo.getUserId());
        entity.setCreatorTime(new Date());
        entity.setLastModifyTime(entity.getCreatorTime());
        entity.setLastModifyUserId(entity.getCreatorUser());
        List<MessageReceiveEntity> receiveEntityList = new ArrayList<>();
        for (String item : toUserIds) {
            MessageReceiveEntity messageReceiveEntity = new MessageReceiveEntity();
            messageReceiveEntity.setId(RandomUtil.uuId());
            messageReceiveEntity.setMessageId(entity.getId());
            messageReceiveEntity.setUserId(item);
            messageReceiveEntity.setIsRead(0);
            receiveEntityList.add(messageReceiveEntity);
        }
        this.save(entity);
        for (MessageReceiveEntity messageReceiveEntity : receiveEntityList) {
            messagereceiveService.save(messageReceiveEntity);
        }
        //消息推送 - PC端
        for (int i = 0; i < toUserIds.size(); i++) {
            for (OnlineUserModel item : OnlineUserProvider.getOnlineUserList()) {
                if (toUserIds.get(i).equals(item.getUserId()) && userInfo.getTenantId().equals(item.getTenantId())) {
                    JSONObject map = new JSONObject();
                    map.put("method", "messagePush");
                    map.put("unreadNoticeCount", 1);
                    map.put("messageType", 2);
                    map.put("userId", userInfo.getTenantId());
                    map.put("toUserId", toUserIds);
                    map.put("title", entity.getTitle());
                    map.put("messageDefaultTime", entity.getLastModifyTime() != null ? entity.getLastModifyTime().getTime() : null);
                    OnlineUserProvider.sendMessage(item, map);
                }
            }
        }
    }

    @Override
    public void sentFlowMessage(List<String> toUserIds, MessageTemplateEntity entity, String content) {
        if (entity != null) {
            // 消息标题
            String title = entity.getTitle();
            this.sentMessage(toUserIds, title, content);
        }
    }

    @Override
    public void logoutWebsocketByToken(String token, String userId){
        if(StringUtil.isNotEmpty(token)) {
            UserProvider.removeWebSocketByToken(token.split(","));
        }else{
            UserProvider.removeWebSocketByUser(userId);
        }
    }

}
